package iot.config;

import iot.cassandra.CassLocationDao;
import iot.cassandra.CassRegistrationDao;
import iot.disruptor.DisruptorDispatcher;
import iot.disruptor.DisruptorQueue;
import iot.entity.Location;
import iot.entity.Registration;
import iot.hbase.LocationAsyncDao;
import iot.hbase.LocationDao;
import iot.hbase.RegistrationAsyncDao;
import iot.hbase.RegistrationDao;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * Created by dave on 18-8-15 下午4:38.
 */
@Configuration
public class DisruptorQueueConfig {
    @Autowired(required = false) LocationAsyncDao locationAsyncDao;
    @Autowired(required = false) RegistrationAsyncDao registrationAsyncDao;
    @Autowired(required = false) LocationDao locationDao;
    @Autowired(required = false) RegistrationDao registrationDao;
    @Autowired(required = false) CassLocationDao cassLocationDao;
    @Autowired(required = false) CassRegistrationDao cassRegistrationDao;

    @Bean
    public DisruptorDispatcher queueDispatcher() {
        // 假定每次写入耗时3ms，单线程1s内可写入300次，20线程可写入6000次
        return new DisruptorDispatcher(20);
    }

    @Bean
    public DisruptorQueue<Registration> regQueue(DisruptorDispatcher dispatcher) {
        return new DisruptorQueue<>(Registration.class, 100, 5,
                list -> {
                    if (registrationDao != null) {
                        registrationDao.batchInsert(list);
                    } else if (registrationAsyncDao != null) {
                        registrationAsyncDao.batchInsert(list);
                    } else {
                        cassRegistrationDao.save(list);
                    }
                },
                dispatcher.getExecutor());
    }

    @Bean
    public DisruptorQueue<Location> locQueue(DisruptorDispatcher dispatcher) {
        return new DisruptorQueue<>(Location.class, 100, 5,
                list -> {
                    if (locationDao != null) {
                        locationDao.batchInsert(list);
                    } else if (locationAsyncDao != null) {
                        locationAsyncDao.batchInsert(list);
                    } else {
                        cassLocationDao.save(list);
                    }
                },
                dispatcher.getExecutor());
    }
}
